package cn.me.alphamq.broker.rebalance.strategy.impl;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.common.model.broker.CommitLogTopicModel;
import cn.me.alphamq.broker.rebalance.ConsumerInstance;
import cn.me.alphamq.broker.rebalance.strategy.RebalanceStrategy;

import java.util.List;
import java.util.Random;

/**
 * 随机重平衡策略：如果队列的数量多于或等于消费者的数量，则将先给每个消费者分配一个队列，剩余的队列随机分给消费者
 *
 * @author f
 */
public class RandomRebalance implements RebalanceStrategy {
    @Override
    public void rebalance(String topic, String consumerGroup) {
        CommitLogTopicModel commitLogTopicModel = Container.fetchCommitLogTopicModelMap().get(topic);
        // 队列数
        int queueNumber = commitLogTopicModel.getQueueList().size();

        List<ConsumerInstance> consumerGroupInstanceList = Container.getConsumerInstanceMap()
                .get(topic)
                .get(consumerGroup);
        // 消费者数
        int consumerNumber = consumerGroupInstanceList.size();

        // 恢复消费者实例 queueIdList 的空集合
        for (ConsumerInstance consumerInstance : consumerGroupInstanceList) {
            consumerInstance.restoreQueueIdList();
        }

        if (queueNumber >= consumerNumber) {
            // 队列的数量 >= 消费者的数量
            int i = 0;
            // 先给每个消费者分配一个队列
            for (ConsumerInstance consumerInstance : consumerGroupInstanceList) {
                consumerInstance.getQueueIdList().add(i++);
            }
            // 剩余的队列随机分给消费者
            Random random = new Random();
            for (; i < queueNumber; i++) {
                consumerGroupInstanceList.get(random.nextInt(consumerNumber)).getQueueIdList().add(i);
            }
        } else {
            // i < 队列的数量 < 消费者的数量
            for (int i = 0; i < queueNumber; i++) {
                consumerGroupInstanceList.get(i).getQueueIdList().add(i);
            }
        }
    }
}
